﻿using Confluent.Kafka;
using System;
using Unity.Credentials;

namespace Unity.Kafka
{
    public abstract class UnityKafkaCredentials : IUnityCredentials
    {
        public string Topic { get; set; }
        public string Servers { get; set; }
        public string GroupId { get; set; }
        public string ClientId { get; set; }
        public string SslCertificationAuthority { get; set; }
        public string SslCert { get; set; }
        public string SslKey { get; set; }
        public string SslKeyPass { get; set; }

        public virtual ProducerConfig GetProducerConfiguration()
        {
            return new ProducerConfig
            {
                BootstrapServers = Servers,
                ClientId = ClientId,
                EnableSslCertificateVerification = true,
                SslCaLocation = SslCertificationAuthority,
                SecurityProtocol = SecurityProtocol.Ssl,
                SslCertificateLocation = SslCert,
                SslKeyLocation = SslKey,
                SslKeyPassword = SslKeyPass
            };
        }

        public virtual IProducer<int, byte[]> GetProducer()
        {
            return new ProducerBuilder<int, byte[]>(GetProducerConfiguration()).Build();
        }

        public virtual ConsumerConfig GetConsumerConfiguration()
        {
            return new ConsumerConfig
            {
                BootstrapServers = Servers,
                GroupId = GroupId,
                AutoOffsetReset = AutoOffsetReset.Earliest,
                EnableSslCertificateVerification = true,
                SslCaLocation = SslCertificationAuthority,
                SecurityProtocol = SecurityProtocol.Ssl,
                SslCertificateLocation = SslCert,
                SslKeyLocation = SslKey,
                SslKeyPassword = SslKeyPass
            };
        }

        public IConsumer<Ignore, byte[]> GetConsumer()
        {
            return new ConsumerBuilder<Ignore, byte[]>(GetConsumerConfiguration()).Build();
        }

    }
}
